#ifndef __RTC_STREAM_MANAGER_H
#define __RTC_STREAM_MANAGER_H
#include <string>
#include <mutex>
#include <set>
#include <unordered_map>
#include "base/lock_free_queue.h"
#include <rtc_base/rtc_certificate.h>
#include "ice/port_allocator.h"
#include "base/event_loop.h"
#include "stream/rtc_stream.h"
namespace xrtc
{
    class PushStream;
    class PullStream;
    struct ComMsg
    {
        int msg_type = -1;
        int worker_id = -1;
        uint64_t uid = 0;
        std::string stream_name;
        std::vector<StreamParams> audio_source;
        std::vector<StreamParams> video_source;
    };
    struct PushStreamSource
    {
        PushStreamSource(uint64_t uid, const std::string &stream_name, const std::vector<StreamParams> &audio_source,
                         const std::vector<StreamParams> &video_source)
            : uid(uid), stream_name(stream_name),
              audio_source(audio_source), video_source(video_source) {}
        uint64_t uid;
        std::string stream_name;
        std::vector<StreamParams> audio_source;
        std::vector<StreamParams> video_source;
    };

    struct DeliverPacket
    {
        std::string stream_name;
        webrtc::MediaType media_type;
        std::shared_ptr<RtcPacket> packet;
    };
    // uid->PullStream*
    typedef std::unordered_map<uint64_t, PullStream *> UserStreamMap;
    // stream_name->UserStreamMap*
    typedef std::unordered_map<std::string, UserStreamMap *> PullStreamMap;

    typedef LockFreeQueue<std::shared_ptr<DeliverPacket>> DeliverPacketQueue;
    // key=> worker_id value=>为 workerid准备的RTP数据队列
    typedef std::map<int, DeliverPacketQueue *> ThreadQueueMap;
    class RtcStreamManager : public RtcStreamListener
    {

    public:
        enum
        {
            // 0~99 表示的是worker_id，代表了从这个worker_id 有新的数据包发送过来
            CM_NEW_DELIVER_PACKET_END = 99,
            CM_NEW_PUSH_STREAM,
            CM_REMOVE_PUSH_STREAM,
            CM_NEW_PULL_STREAM,
            CM_REMOVE_PULL_STREAM,

        };
        RtcStreamManager(EventLoop *el);
        ~RtcStreamManager();
        int Init();
        void processNotify(int type);
        int Notify(int type);
        int GetWorkerId();
        int NotifyMsg(std::shared_ptr<ComMsg> msg);
        void NotifyRemovePushStream(PushStream *push_stream);
        void NotifyNewPullStream(PullStream *pull_stream);
        void NotifyNewPushStream(PushStream *push_stream);
        void NotifyNewPacket(int worker_id, std::shared_ptr<DeliverPacket> packet);
        void NotifyRemovePullStream(const std::string &stream_name);

        // 根据参数生成offer
        int CreatePushStream(uint64_t uid,
                             const std::string &stream_name,
                             bool audio,
                             bool video,
                             bool is_dtls,
                             bool is_pli,
                             const std::string &mode,
                             uint32_t log_id,
                             rtc::RTCCertificate *certificate,
                             std::string &offer);
        int CreatePullStream(uint64_t uid,
                             const std::string &stream_name,
                             bool audio,
                             bool video,
                             bool is_dtls,
                             const std::string &mode,
                             uint32_t log_id,
                             rtc::RTCCertificate *certificate,
                             std::string &offer);
        int SetAnswer(uint64_t uid, const std::string &stream_name, std::string &answer, std::string &stream_type, const std::string &mode, uint32_t log_id);

        int StopPush(uint64_t uid, const std::string &stream_name);
        int StopPull(uint64_t uid, const std::string &stream_name, const std::string &mode);

        void OnConnectionState(RtcStream *stream, PeerConnectionState state) override;
        // transport
        void OnRtpPacketReceived(RtcStream *stream, const char *data, size_t len) override;
        void OnRtcpPacketReceived(RtcStream *stream, const char *data, size_t len) override;
        // live
        void OnRtpPacket(RtcStream *stream, webrtc::MediaType media_type, std::shared_ptr<RtcPacket> packet) override;
        void OnStreamException(RtcStream *stream) override;
        void OnSrInfo(RtcStream *stream, webrtc::MediaType media_type, uint32_t rtp_timestamp,
                      webrtc::NtpTime ntp) override;
        void OnNackReceived(RtcStream *, webrtc::MediaType media_type, const std::vector<uint16_t> &nack_list) override;

    private:
        void RemovePushStream(RtcStream *stream);
        void RemovePushStream(uint64_t uid, const std::string &stream_name);
        PushStream *FindPushStream(const std::string &stream_name);
        void RemovePullStream(RtcStream *stream);
        void RemovePullStream(uint64_t uid, const std::string &stream_name);
        PullStream *FindPullStream(const std::string &stream_name);

        void AddPullStreamM(PullStream *stream);
        void RemovePullStreamM(RtcStream *stream);
        void RemovePullStreamM(uint64_t uid, const std::string &stream_name);
        PullStream *FindPullStreamM(uint64_t uid, const std::string &stream_name);
        void ProcessNewPushStream(std::shared_ptr<ComMsg> msg);
        void ProcessRemovePushStream(std::shared_ptr<ComMsg> msg);
        void ProcessNewPullStream(std::shared_ptr<ComMsg> msg);
        void DeliverNewPacket(const std::string &stream_name, webrtc::MediaType stream_type, std::shared_ptr<RtcPacket> packet);
        void ProcessNewDeliverPacket(int worker_id);
        void ProcessRemovePullStream(std::shared_ptr<ComMsg> msg);

        std::shared_ptr<PushStreamSource> FindPushStreamSource(const std::string &stream_name);
        DeliverPacketQueue *FindQueueByWorkerId(int worker_id);
        void SendPacket(const std::string &stream_name, webrtc::MediaType media_type, std::shared_ptr<RtcPacket> packet);

    private:
        EventLoop *el_;
        IOWatcher *pipe_watcher_ = nullptr;
        int notify_send_fd_ = 0;
        int notify_recv_fd_ = 0;
        LockFreeQueue<std::shared_ptr<ComMsg>> mq_;
        std::mutex mq_mutex_;
        std::unordered_map<std::string, PushStream *> push_streams_; // 推流信息
        // transparent 拉流
        std::unordered_map<std::string, PullStream *> pull_streams_; // 拉流信息
        // 1v多拉流 Live
        PullStreamMap multi_pull_streams_;

        std::unique_ptr<PortAllocator> allocator_;
        // map  保存了其他worker 当中的推流源相关的信息
        std::unordered_map<std::string, std::shared_ptr<PushStreamSource>> push_stream_sources_;
        // map  保存了订阅源流所有的worker_id
        std::unordered_map<std::string, std::set<int>> pull_workers_;
        ThreadQueueMap queue_map_;
    };

} // namespace xrtc

#endif